package com.aysafety.core.DBDriver.hbase;


import com.aysafety.core.common.DateUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.client.coprocessor.AggregationClient;
import org.apache.hadoop.hbase.client.coprocessor.LongColumnInterpreter;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.math.BigDecimal;
import java.util.*;

public class HBaseDriver {
    private static Configuration conf = null;
    private static Connection conn = null;
    private static int cache = 1000;

    public static void setZKConfig(String zkHost, String zkPort) throws Exception {
        conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", zkHost);
        conf.set("hbase.zookeeper.property.clientPort", zkPort);
        conf.setInt("hbase.client.scanner.timeout.period", 300000);
    }

    public static void connection() throws IOException {
        if (null == conn) {
            conn = ConnectionFactory.createConnection(conf);
        }
    }

    public static void close() throws IOException {
        if (null != conn) {
            conn.close();
        }
    }

    public static boolean isConn() {
        if (null != conn) {
            return true;
        }
        return false;
    }

    // 创建数据表
    public static boolean createTable(String tableName, String[] columnFamilys) throws IOException {
        // 获取管理员权限
        HBaseAdmin hAdmin = (HBaseAdmin) conn.getAdmin();
        if (isExistTable(tableName)) {
            return false;
        } else {
            // 新建一个表描述
            HTableDescriptor tableDesc = new HTableDescriptor(TableName.valueOf(tableName));
            tableDesc.addCoprocessor("org.apache.hadoop.hbase.coprocessor.AggregateImplementation");
            // 在表描述里添加列族
            for (String columnFamily : columnFamilys) {
                tableDesc.addFamily(new HColumnDescriptor(columnFamily));
            }
            // 根据配置好的表描述建表
            hAdmin.createTable(tableDesc);

        }
        return true;
    }

    // 检查数据表是否存在
    public static boolean isExistTable(String tableName) throws IOException {
        HBaseAdmin hAdmin = (HBaseAdmin) conn.getAdmin();
        if (hAdmin.tableExists(tableName)) {
            return true;
        }
        return false;
    }

    // 删除数据表
    public static boolean deleteTable(String tableName) throws IOException {
        // 创建一个数据库管理员
        HBaseAdmin hAdmin = (HBaseAdmin) conn.getAdmin();
        if (isExistTable(tableName)) {
            // 失效表
            hAdmin.disableTable(tableName);
            // 删除表
            hAdmin.deleteTable(tableName);
            return true;
        }
        return false;
    }

    // 增加一条数据
    public static String addRow(String tableName, String rowKey, String columnFamily, String column, String value)
        throws IOException {
        // 获取表
        HTable table = (HTable) conn.getTable(TableName.valueOf(tableName));
        // 通过rowkey创建一个put对象
        Put put = new Put(Bytes.toBytes(rowKey));
        // 在put对象中设置列族、列、值
        put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column), Bytes.toBytes(value));
        // 插入数据,可通过put(List<Put>)批量插入
        table.put(put);
        // 关闭资源
        table.close();
        return rowKey;
    }

    public static String addRow(String tableName, String rowKey, String columnFamily, String column, double value)
        throws IOException {
        // 获取表
        HTable table = (HTable) conn.getTable(TableName.valueOf(tableName));
        // 通过rowkey创建一个put对象
        Put put = new Put(Bytes.toBytes(rowKey));
        // 在put对象中设置列族、列、值
        put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column), Bytes.toBytes(value));
        // 插入数据,可通过put(List<Put>)批量插入
        table.put(put);
        // 关闭资源
        table.close();
        return rowKey;
    }

    public static String addRow(String tableName, String rowKey, String columnFamily, String column, float value)
        throws IOException {
        // 获取表
        HTable table = (HTable) conn.getTable(TableName.valueOf(tableName));
        // 通过rowkey创建一个put对象
        Put put = new Put(Bytes.toBytes(rowKey));
        // 在put对象中设置列族、列、值
        put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column), Bytes.toBytes(value));
        // 插入数据,可通过put(List<Put>)批量插入
        table.put(put);
        // 关闭资源
        table.close();
        return rowKey;
    }


    public static String addRow(String tableName, String rowKey, String columnFamily, String column, int value)
        throws IOException {
        // 获取表
        HTable table = (HTable) conn.getTable(TableName.valueOf(tableName));
        // 通过rowkey创建一个put对象
        Put put = new Put(Bytes.toBytes(rowKey));
        // 在put对象中设置列族、列、值
        put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column), Bytes.toBytes(value));
        // 插入数据,可通过put(List<Put>)批量插入
        table.put(put);
        // 关闭资源
        table.close();
        return rowKey;
    }

    public static String addRow(String tableName, String rowKey, String columnFamily, String column, BigDecimal value)
        throws IOException {
        // 获取表
        HTable table = (HTable) conn.getTable(TableName.valueOf(tableName));
        // 通过rowkey创建一个put对象
        Put put = new Put(Bytes.toBytes(rowKey));
        // 在put对象中设置列族、列、值
        put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column), Bytes.toBytes(value));
        // 插入数据,可通过put(List<Put>)批量插入
        table.put(put);
        // 关闭资源
        table.close();
        return rowKey;
    }

    public static String addRow(String tableName, String rowKey, String columnFamily, String column, long value)
        throws IOException {
        // 获取表
        HTable table = (HTable) conn.getTable(TableName.valueOf(tableName));
        // 通过rowkey创建一个put对象
        Put put = new Put(Bytes.toBytes(rowKey));
        // 在put对象中设置列族、列、值
        put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column), Bytes.toBytes(value));
        // 插入数据,可通过put(List<Put>)批量插入
        table.put(put);
        // 关闭资源
        table.close();
        return rowKey;
    }


    public static String addRow(String tableName, String rowKey, String columnFamily, String column, boolean value)
        throws IOException {
        // 获取表
        HTable table = (HTable) conn.getTable(TableName.valueOf(tableName));
        // 通过rowkey创建一个put对象
        Put put = new Put(Bytes.toBytes(rowKey));
        // 在put对象中设置列族、列、值
        put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column), Bytes.toBytes(value));
        // 插入数据,可通过put(List<Put>)批量插入
        table.put(put);
        // 关闭资源
        table.close();
        return rowKey;
    }

    public static String addRow(String tableName, String rowKey, String columnFamily, String column, short value)
        throws IOException {
        // 获取表
        HTable table = (HTable) conn.getTable(TableName.valueOf(tableName));
        // 通过rowkey创建一个put对象
        Put put = new Put(Bytes.toBytes(rowKey));
        // 在put对象中设置列族、列、值
        put.addColumn(Bytes.toBytes(columnFamily), Bytes.toBytes(column), Bytes.toBytes(value));
        // 插入数据,可通过put(List<Put>)批量插入
        table.put(put);
        // 关闭资源
        table.close();
        return rowKey;
    }

    public static boolean isExistRowByRowKey(String tableName, String rowKey) throws IOException {
        HTable table = (HTable) conn.getTable(TableName.valueOf(tableName));
        Get getData = new Get(rowKey.getBytes());
        Result result = table.get(getData);
        table.close();
        if (null == result.listCells()) {
            return false;
        }
        return true;
    }


    // 删除一条数据
    public static boolean delRow(String tableName, String rowKey) throws IOException {
        // 获取表
        HTable table = (HTable) conn.getTable(TableName.valueOf(tableName));
        // 删除数据
        Delete delete = new Delete(Bytes.toBytes(rowKey));
        table.delete(delete);
        // 关闭资源
        table.close();
        return true;
    }

    public static List<Map<String, Object>> queryData(String tableName, List<Map<String, Object>> conditions, int page,
        int size, Map<String, Object> structure, String startRow, String endRow) throws Throwable {
        if (!isConn()) {
            connection();
        }
        List<Map<String, Object>> resultList = new ArrayList<Map<String, Object>>();
        HTable table = (HTable) conn.getTable(TableName.valueOf(tableName));
        Scan scan = new Scan();
        scan.setCaching(cache);
        scan.setCacheBlocks(false);
        scan.setReversed(true);
        if (StringUtils.isNotBlank(startRow)) {
            scan.withStartRow(startRow.getBytes());
        }
        if (StringUtils.isNotBlank(endRow)) {
            scan.withStopRow(endRow.getBytes());
        }
        FilterList filterList = new FilterList(FilterList.Operator.MUST_PASS_ALL);
        for (Map<String, Object> condition : conditions) {
            SingleColumnValueFilter filter = parseCondition(condition);
            filterList.addFilter(filter);
        }
        Filter pageFilter = new PageFilter(size);
        filterList.addFilter(pageFilter);
        scan.setFilter(filterList);
        ResultScanner res = table.getScanner(scan);
        for (Result result : res) {
            resultList.add(getRowByResult(result, structure));
        }
        res.close();
        table.close();
        return resultList;
    }

    public static long countData(String tableName) throws Throwable {
        if (!isConn()) {
            connection();
        }
        long count = 0;
        // 获取表
        HTable table = (HTable) conn.getTable(TableName.valueOf(tableName));
        Scan scan = new Scan();
        scan.setCaching(cache);
        scan.setCacheBlocks(false);
        AggregationClient aggregation = new AggregationClient(conf);
        Long result = aggregation.rowCount(table, new LongColumnInterpreter(), scan);
        count = result.intValue();
        aggregation.close();
        table.close();
        return count;
    }

    public static long countData(String tableName, List<Map<String, Object>> conditions, String startRow, String endRow)
        throws Throwable {
        if (!isConn()) {
            connection();
        }
        long count = 0;
        // 获取表
        HTable table = (HTable) conn.getTable(TableName.valueOf(tableName));
        Scan scan = new Scan();
        scan.setCaching(cache);
        scan.setCacheBlocks(false);
        scan.setReversed(true);
        if (StringUtils.isNotBlank(startRow)) {
            scan.withStartRow(startRow.getBytes());
        }
        if (StringUtils.isNotBlank(endRow)) {
            scan.withStopRow(endRow.getBytes());
        }
        FilterList filterList = new FilterList(FilterList.Operator.MUST_PASS_ALL);
        for (Map<String, Object> condition : conditions) {
            SingleColumnValueFilter filter = parseCondition(condition);
            filterList.addFilter(filter);
        }
        scan.setFilter(filterList);
        ResultScanner res = table.getScanner(scan);
        for (Result result : res) {
            count++;
        }
        res.close();
        table.close();
        return count;
    }

    private static SingleColumnValueFilter parseCondition(Map<String, Object> condition) throws Throwable {
        SingleColumnValueFilter filter = null;
        CompareFilter.CompareOp compareOp = null;
        if (condition.get("symbol").toString().equals("LESS")) {
            compareOp = CompareFilter.CompareOp.LESS;
        } else if (condition.get("symbol").toString().equals("LESS_OR_EQUAL")) {
            compareOp = CompareFilter.CompareOp.LESS_OR_EQUAL;
        } else if (condition.get("symbol").toString().equals("GREATER")) {
            compareOp = CompareFilter.CompareOp.GREATER;
        } else if (condition.get("symbol").toString().equals("GREATER_OR_EQUAL")) {
            compareOp = CompareFilter.CompareOp.GREATER_OR_EQUAL;
        } else if (condition.get("symbol").toString().equals("NO_OP")) {
            compareOp = CompareFilter.CompareOp.NO_OP;
        } else if (condition.get("symbol").toString().equals("NOT_EQUAL")) {
            compareOp = CompareFilter.CompareOp.NOT_EQUAL;
        } else {
            compareOp = CompareFilter.CompareOp.EQUAL;
        }
        if (condition.get("valueType").toString().equals("Long")) {
            filter = new SingleColumnValueFilter(Bytes.toBytes(condition.get("columnFamily").toString()),
                Bytes.toBytes(condition.get("columnId").toString()), compareOp,
                Bytes.toBytes(Long.valueOf(condition.get("value").toString())));
        } else if (condition.get("valueType").toString().equals("Short")) {
            filter = new SingleColumnValueFilter(Bytes.toBytes(condition.get("columnFamily").toString()),
                Bytes.toBytes(condition.get("columnId").toString()), compareOp,
                Bytes.toBytes(Short.valueOf(condition.get("value").toString())));
        } else if (condition.get("valueType").toString().equals("Integer")) {
            filter = new SingleColumnValueFilter(Bytes.toBytes(condition.get("columnFamily").toString()),
                Bytes.toBytes(condition.get("columnId").toString()), compareOp,
                Bytes.toBytes(Integer.valueOf(condition.get("value").toString())));
        } else if (condition.get("valueType").toString().equals("Boolean")) {
            filter = new SingleColumnValueFilter(Bytes.toBytes(condition.get("columnFamily").toString()),
                Bytes.toBytes(condition.get("columnId").toString()), compareOp,
                Bytes.toBytes(Boolean.valueOf(condition.get("value").toString())));
        } else if (condition.get("valueType").toString().equals("BigDecimalLong")) {
            filter = new SingleColumnValueFilter(Bytes.toBytes(condition.get("columnFamily").toString()),
                Bytes.toBytes(condition.get("columnId").toString()), compareOp,
                Bytes.toBytes(BigDecimal.valueOf(Long.valueOf(condition.get("value").toString()))));
        } else if (condition.get("valueType").toString().equals("BigDecimalDouble")) {
            filter = new SingleColumnValueFilter(Bytes.toBytes(condition.get("columnFamily").toString()),
                Bytes.toBytes(condition.get("columnId").toString()), compareOp,
                Bytes.toBytes(BigDecimal.valueOf(Double.valueOf(condition.get("value").toString()))));
        } else if (condition.get("valueType").toString().equals("Double")) {
            filter = new SingleColumnValueFilter(Bytes.toBytes(condition.get("columnFamily").toString()),
                Bytes.toBytes(condition.get("columnId").toString()), compareOp,
                Bytes.toBytes(Double.valueOf(condition.get("value").toString())));
        } else if (condition.get("valueType").toString().equals("Float")) {
            filter = new SingleColumnValueFilter(Bytes.toBytes(condition.get("columnFamily").toString()),
                Bytes.toBytes(condition.get("columnId").toString()), compareOp,
                Bytes.toBytes(Float.valueOf(condition.get("value").toString())));
        } else {
            filter = new SingleColumnValueFilter(Bytes.toBytes(condition.get("columnFamily").toString()),
                Bytes.toBytes(condition.get("columnId").toString()), compareOp,
                Bytes.toBytes(condition.get("value").toString()));
        }
        filter.setFilterIfMissing(true);
        return filter;
    }

    private static Map<String, Object> getRowByResult(Result result, Map<String, Object> structure) {
        if (result == null) {
            return null;
        }
        Map<String, Object> cellMap = new HashMap<String, Object>();
        for (Cell cell : result.listCells()) {
            String rowkey = Bytes.toString(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength());
            String cf = Bytes.toString(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength());
            String qf = Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());
            if (structure.containsKey(qf)) {
                String structureType = structure.get(qf).toString();
                if (structureType.equals("Date")) {
                    if(cell.getValueLength()>0){
                        Long timestamp = Bytes.toLong(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
                        Date dateTime = DateUtils.fomatDateTime(timestamp);
                        cellMap.put(qf, DateUtils.getTime(dateTime));
                    }else{
                        cellMap.put(qf, Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength()));
                    }

                } else if (structureType.equals("Short")) {
                    cellMap.put(qf, Bytes.toShort(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength()));
                } else if (structureType.equals("Long")) {
                    cellMap.put(qf, Bytes.toLong(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength()));
                } else if (structureType.equals("Integer")) {
                    cellMap.put(qf, Bytes.toInt(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength()));
                } else if (structureType.equals("BigDecimal")) {
                    cellMap.put(qf,
                        Bytes.toBigDecimal(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength()));
                } else if (structureType.equals("Boolean")) {
                    byte[] temp = cell.getValueArray();
                    byte valueTemp = temp[temp.length - 1];
                    cellMap.put(qf, Bytes.toBoolean(new byte[] {valueTemp}));
                } else if (structureType.equals("Double")) {
                    cellMap.put(qf, Bytes.toDouble(cell.getValueArray(), cell.getValueOffset()));
                } else if (structureType.equals("Float")) {
                    cellMap.put(qf, Bytes.toFloat(cell.getValueArray(), cell.getValueOffset()));
                } else {
                    cellMap.put(qf, Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength()));
                }
            } else {
                cellMap.put(qf, Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength()));
            }
            cellMap.put("rowkey", rowkey);
            cellMap.put("columnFamily", cf);

        }
        return cellMap;
    }

}
